package javaVersion.learn.batchProcess;

import org.apache.commons.io.FileUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.configuration.Configuration;

import java.io.File;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

public class Pro8_distributeCache {
    public static void main(String[] args) throws Exception {
        //获取环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //加载本地集合
        DataSource<String> source = env.fromElements("a", "b", "c", "d");
        //注册分布式缓存文件
        env.registerCachedFile("./data/MyFlinkFile.txt","a_txt");
        //map转换数据
        MapOperator<String, String> cacheSource = source.map(new RichMapFunction<String, String>() {
            ArrayList<String> dataList = new ArrayList<String>();
            @Override
            public void open(Configuration parameters) throws Exception {
                //读取缓存文件
                File myFile = getRuntimeContext().getDistributedCache().getFile("a_txt");
                List<String> lines = FileUtils.readLines(myFile);
                Iterator<String> it = lines.iterator();
                while (it.hasNext()) {
                    String str = it.next();
                    System.out.println(str);
                    this.dataList.add(str);
                }
            }

            @Override
            public String map(String value) {
                //map中各个数据使用分布式缓存数据
                ArrayList<String> resList = new ArrayList<>();
                for (String s : dataList) {
                    String res = value + s;
                    resList.add(res);
                }
                return "原map里数据:"+ value + ",分布式缓存数据:" + dataList + ",结合后的数据: "+resList;
            }
        });
        cacheSource.print();
    }
}
